package com.example.kafka.exception.poison;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 处理毒丸消息的Service
 */
@Service
public class PoisonMessageService {
    private final Logger log = LoggerFactory.getLogger(getClass());

    // 用于记录消息重试次数
    private final ConcurrentHashMap<String, AtomicInteger> retryCountMap = new ConcurrentHashMap<>();
    private static final int MAX_RETRY_COUNT = 3;

    /**
     * 用例1: 处理格式错误的消息
     * 模拟JSON格式错误的情况
     */
    @KafkaListener(topics = "invalid-format-topic", groupId = "invalid-format-group")
    public String handleInvalidFormat(String message, Acknowledgment ack) {
        String messageId = extractMessageId(message);
        int retryCount = getRetryCount(messageId);

        try {
            if (message.contains("invalid-json")) {
                throw new IllegalArgumentException("Invalid JSON format");
            }
            return "Successfully processed message: " + message;
        } catch (IllegalArgumentException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                incrementRetryCount(messageId);
                log.warn("Retrying invalid format message: {}, attempt: {}", messageId, retryCount + 1);
                throw e; // 重新抛出异常以触发重试
            } else {
                log.error("Message {} exceeded max retry count, moving to DLQ", messageId);
                moveToDeadLetterQueue(messageId, message, e.getMessage());
                ack.acknowledge(); // 确认消息以避免重复处理
                return "Message moved to DLQ due to invalid format";
            }
        }
    }

    /**
     * 用例2: 处理业务规则验证失败
     * 模拟业务规则验证失败的情况
     */
    @KafkaListener(topics = "business-rule-topic", groupId = "business-rule-group")
    public String handleBusinessRuleViolation(String message, Acknowledgment ack) {
        String messageId = extractMessageId(message);
        int retryCount = getRetryCount(messageId);

        try {
            if (message.contains("invalid-business-rule")) {
                throw new IllegalStateException("Business rule violation");
            }
            return "Successfully processed message: " + message;
        } catch (IllegalStateException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                incrementRetryCount(messageId);
                log.warn("Retrying business rule violation message: {}, attempt: {}", messageId, retryCount + 1);
                throw e;
            } else {
                log.error("Message {} exceeded max retry count, moving to DLQ", messageId);
                moveToDeadLetterQueue(messageId, message, e.getMessage());
                ack.acknowledge();
                return "Message moved to DLQ due to business rule violation";
            }
        }
    }

    /**
     * 用例3: 处理依赖服务不可用
     * 模拟外部服务不可用的情况
     */
    @KafkaListener(topics = "dependency-topic", groupId = "dependency-group")
    public String handleDependencyUnavailable(String message, Acknowledgment ack) {
        String messageId = extractMessageId(message);
        int retryCount = getRetryCount(messageId);

        try {
            if (message.contains("service-unavailable")) {
                throw new RuntimeException("Dependency service unavailable");
            }
            return "Successfully processed message: " + message;
        } catch (RuntimeException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                incrementRetryCount(messageId);
                log.warn("Retrying dependency unavailable message: {}, attempt: {}", messageId, retryCount + 1);
                throw e;
            } else {
                log.error("Message {} exceeded max retry count, moving to DLQ", messageId);
                moveToDeadLetterQueue(messageId, message, e.getMessage());
                ack.acknowledge();
                return "Message moved to DLQ due to dependency unavailability";
            }
        }
    }

    /**
     * 用例4: 处理数据一致性错误
     * 模拟数据不一致的情况
     */
    @KafkaListener(topics = "data-consistency-topic", groupId = "data-consistency-group")
    public String handleDataInconsistency(String message, Acknowledgment ack) {
        String messageId = extractMessageId(message);
        int retryCount = getRetryCount(messageId);

        try {
            if (message.contains("inconsistent-data")) {
                throw new IllegalStateException("Data consistency violation");
            }
            return "Successfully processed message: " + message;
        } catch (IllegalStateException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                incrementRetryCount(messageId);
                log.warn("Retrying data inconsistency message: {}, attempt: {}", messageId, retryCount + 1);
                throw e;
            } else {
                log.error("Message {} exceeded max retry count, moving to DLQ", messageId);
                moveToDeadLetterQueue(messageId, message, e.getMessage());
                ack.acknowledge();
                return "Message moved to DLQ due to data inconsistency";
            }
        }
    }

    /**
     * 用例5: 处理权限验证失败
     * 模拟权限验证失败的情况
     */
    @KafkaListener(topics = "authorization-topic", groupId = "authorization-group")
    public String handleAuthorizationFailure(String message, Acknowledgment ack) {
        String messageId = extractMessageId(message);
        int retryCount = getRetryCount(messageId);

        try {
            if (message.contains("unauthorized")) {
                throw new SecurityException("Authorization failed");
            }
            return "Successfully processed message: " + message;
        } catch (SecurityException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                incrementRetryCount(messageId);
                log.warn("Retrying authorization failure message: {}, attempt: {}", messageId, retryCount + 1);
                throw e;
            } else {
                log.error("Message {} exceeded max retry count, moving to DLQ", messageId);
                moveToDeadLetterQueue(messageId, message, e.getMessage());
                ack.acknowledge();
                return "Message moved to DLQ due to authorization failure";
            }
        }
    }

    // 辅助方法
    private String extractMessageId(String message) {
        // 实际应用中应该从消息中提取唯一标识符
        return message.hashCode() + "";
    }

    private int getRetryCount(String messageId) {
        return retryCountMap.computeIfAbsent(messageId, k -> new AtomicInteger(0)).get();
    }

    private void incrementRetryCount(String messageId) {
        retryCountMap.computeIfAbsent(messageId, k -> new AtomicInteger(0)).incrementAndGet();
    }

    private void moveToDeadLetterQueue(String messageId, String message, String error) {
        // 实际应用中应该将消息发送到死信队列
        log.info("Moving message {} to DLQ with error: {}", messageId, error);
    }
}
